package com.example.day04.controller.sideoutputs;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.*;

public class WaterDemo3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        List<WaterEntityDemo> list = Arrays.asList(
                new WaterEntityDemo("s1", 1L, 1),
                new WaterEntityDemo("s2", 1L, 1),
                new WaterEntityDemo("s3", 2L, 1),
                new WaterEntityDemo("s3", 1L, 1), // 注意：id 为空，因为第二个元素缺少 id
                new WaterEntityDemo("s2", 1L, 1),
                new WaterEntityDemo("s2", 12L, 1),
                new WaterEntityDemo("s2", 13L, 1),
                new WaterEntityDemo("s2", 14L, 1),
                new WaterEntityDemo("s1", 1L, 1),
                new WaterEntityDemo("s1", 2L, 1),
                new WaterEntityDemo("s1", 3L, 1),
                new WaterEntityDemo("s1", 5L, 1),
                new WaterEntityDemo("s1", 9L, 1),
                new WaterEntityDemo("s1", 10L, 1),
                new WaterEntityDemo("s1", 12L, 2),
                new WaterEntityDemo("s2", 12L, 3),
                new WaterEntityDemo("s1", 16L, 22),
                new WaterEntityDemo("s2", 16L, 33),
                new WaterEntityDemo("s3", 14L, 1),
                new WaterEntityDemo("s2", 16L, 33),
                new WaterEntityDemo("s3", 16L, 2),
                new WaterEntityDemo("s3", 20L, 1)
        );

        WaterEntityDemo[] array = list.toArray(new WaterEntityDemo[0]);

        SingleOutputStreamOperator<WaterEntityDemo> waterEntitySingleOutputStreamOperator = env.fromElements(array).assignTimestampsAndWatermarks(WatermarkStrategy.<WaterEntityDemo>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner((element, recordTimestamp) -> element.getVc() * 1000L));


        SingleOutputStreamOperator<String> process = waterEntitySingleOutputStreamOperator.windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .process(new MyTopNpAWF());

        process.print();
//        Top-1: id=s1, count=4窗口的开始时间结束时间：-5000 - 5000
//        Top-2: id=s3, count=2窗口的开始时间结束时间：-5000 - 5000
//        Top-3: id=s2, count=2窗口的开始时间结束时间：-5000 - 5000
//        Top-1: id=s1, count=6窗口的开始时间结束时间：0 - 10000
//        Top-2: id=s3, count=2窗口的开始时间结束时间：0 - 10000
//        Top-3: id=s2, count=2窗口的开始时间结束时间：0 - 10000
//        Top-1: id=s1, count=4窗口的开始时间结束时间：5000 - 15000
//        Top-2: id=s2, count=4窗口的开始时间结束时间：5000 - 15000
//        Top-3: id=s3, count=1窗口的开始时间结束时间：5000 - 15000

        env.execute("aaa");
    }
    public static class MyTopNpAWF extends ProcessAllWindowFunction<WaterEntityDemo, String, TimeWindow> {
        @Override
        public void process(Context context, Iterable<WaterEntityDemo> elements, Collector<String> out) throws Exception {
            Map<String, Integer> map = new HashMap<>();

            for (WaterEntityDemo element : elements) {
                String id = element.getId();
                if (!map.containsKey(id)) {
                    map.put(id, 1);
                } else {
                    map.put(id, map.get(id) + 1);
                }
            }

            List<Tuple2<String, Integer>> datas = new ArrayList<>();
            for (String key : map.keySet()) {
                datas.add(Tuple2.of(key, map.get(key)));
            }

            // 降序排列
            datas.sort((o1, o2) -> o2.f1.compareTo(o1.f1)); // 使用 compareTo 而不是减法

            // 输出 Top 3 结果
            int topN = 3;
            for (int i = 0; i < Math.min(datas.size(), topN); i++) {
                Tuple2<String, Integer> data = datas.get(i);
                out.collect("Top-" + (i + 1) + ": id=" + data.f0 + ", count=" + data.f1 + "窗口的开始时间结束时间：" + context.window().getStart() + " - " + context.window().getEnd());
            }
        }
    }
}
